package com.grwx.omc.kafka.basekafka.handle;

import com.grwx.omc.common.constant.KafkaConstant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @Description: 消息服务的测试监听
 * @Param:
 * @return:
 * @Author: ManJiang
 * @Date: 2020/5/8
 */
@Component
public class MessageHandle {
    private static Logger logger = LoggerFactory.getLogger(MessageHandle.class);

    @KafkaListener(topics = KafkaConstant.TOPIC_TEST,containerFactory = "ackContainerFactory")
    public void handleMessage(ConsumerRecord record, Acknowledgment acknowledgment){
        try {
            String message = (String) record.value();
            logger.info("收到消息: {}", message);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            // 手动回复
            acknowledgment.acknowledge();
        }
    }

}
